package zhongbo.java.air.web.websocket;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;
import zhongbo.java.air.config.RedisListenerConfig;
import zhongbo.java.air.listener.RedisLogChaneIPublisher;
import zhongbo.java.air.listener.RedisSubscribeLogChannelListener;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

@Slf4j
@Component
public class LiveWebSocket extends TextWebSocketHandler {

    private final List<WebSocketSession> sessions = new CopyOnWriteArrayList<>();
    private final RedisLogChaneIPublisher redisLogChaneIPublisher;
    private final RedisSubscribeLogChannelListener redisSubscribeLogChannelListener;

    @Autowired
    public LiveWebSocket(RedisLogChaneIPublisher redisLogChaneIPublisher, RedisSubscribeLogChannelListener redisSubscribeLogChannelListener) {
        this.redisLogChaneIPublisher = redisLogChaneIPublisher;
        this.redisSubscribeLogChannelListener = redisSubscribeLogChannelListener;
        // 订阅 Redis 频道
        redisSubscribeLogChannelListener.subscribe(RedisListenerConfig.INST_FLIGHT_PATH, this::broadcastMessage);
    }

    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        log.info("WebSocket 连接已建立: {}", session.getId());
        sessions.add(session);
        super.afterConnectionEstablished(session);
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        log.info("WebSocket 连接已关闭: {}", session.getId());
        sessions.remove(session);
        super.afterConnectionClosed(session, status);
    }

    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        log.info("收到消息: {}", message.getPayload());
        // 处理接收到的消息并返回响应
        String response = "已收到消息: " + message.getPayload();
        session.sendMessage(new TextMessage(response));
    }

    private void broadcastMessage(String message) {
        log.info("广播消息: {}", message);
        for (WebSocketSession session : sessions) {
            if (!session.isOpen()) {
                continue;
            }
            synchronized (session) {
                try {
                    session.sendMessage(new TextMessage(message));
                    log.debug("消息已发送到会话: {}", session.getId());
                } catch (IOException e) {
                    log.error("广播消息时发生错误: {}", e.getMessage());
                }
            }
        }
    }
}